package com.black.common.rabbitmq.producer;

import com.black.common.util.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author linwenfeng
 */
@Component
@Slf4j
public class RabbitSender {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    JdbcTemplate jdbcTemplate;

    @PostConstruct
    public void reCallBack(){
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);
    }

    /** 消息发送到交换机回调
     * @param correlationData 消息的附加信息
     * @param ack 是否接受
     * @param cause 拒收原因
     */
   final ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
        assert correlationData != null;
        if (ack){
           //更新消息状态--完成
            jdbcTemplate.update("update messageSender set status = 1,updatedDate = now() where id = ? ",correlationData.getId());
       }else{
           //更新消息状态--失败
            jdbcTemplate.update("update messageSender set status = 2,errorMessage = ? ,updatedDate = now() where id =  ? ",cause,correlationData.getId());
            log.error("发生MQ消息失败:{}",cause);
       }
   };

    /**
     * 交换机转发消息到队列回调
     * @param message 被退回的消息
     * @param replyCode 错误代码
     * @param replyText 错误描述
     * @param exchange 交换机
     * @param routingKey 路由器键
     */
    final ReturnCallback returnCallback = (message, replyCode, replyText, exchange, routingKey) -> jdbcTemplate.update("update messageSender set status = '2',errorMessage = ? ,updatedDate = now() where id =  ? ",replyText,message.getMessageProperties().getMessageId());

    /***
     * 发送MQ消息
     * @param message 消息体
     * @param messageId 消息ID
     * @param exchange 交换机
     * @param routingKey 路由键
     */
    public void sendMessage(Object message, String messageId, String exchange, String routingKey) {
        saveMessage(message, messageId, exchange, routingKey);
        CorrelationData correlationData = new CorrelationData(messageId);
        rabbitTemplate.convertAndSend(exchange,routingKey,message,correlationData);
    }

    /***
     * 保存MQ消息
     * @param message 消息体
     * @param messageId 消息ID
     * @param exchange 交换机
     * @param routingKey 路由键
     */
    private void saveMessage(Object message, String messageId, String exchange, String routingKey) {
        String messagePayload = JsonUtils.toJson(message);
        jdbcTemplate.update("insert into messageSender(id,exchange,routingKey,payload) values(?,?,?,?)",messageId,exchange,routingKey,messagePayload);
    }

}
